package com.csw.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.connector.jdbc.{JdbcConnectionOptions, JdbcSink, JdbcStatementBuilder}
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

object Demo02MysqlSink {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)
    val ds: DataStream[String] = env.readTextFile("flink/data/students.txt")

    //使用自定义的sink
//    ds.addSink(new MysqlSink)

    //使用官网的jdbc

    ds.addSink(JdbcSink.sink(
      "insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)",
      new JdbcStatementBuilder[String] {
        override def accept(t: PreparedStatement, u: String): Unit = {
          val split: Array[String] = u.split(",")
          val id: Int = split(0).toInt
          val name: String = split(1)
          val age: Int = split(2).toInt
          val gender: String = split(3)
          val clazz: String = split(4)

          t.setInt(1,id)
          t.setString(2, name)
          t.setInt(3, age)
          t.setString(4, gender)
          t.setString(5, clazz)
        }
      }
      ,
      new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://master:3306/csw?useUnicode=true&characterEncoding=utf-8&useSSL=false")
        .withDriverName("com.mysql.jdbc.Driver")
        .withUsername("root")
        .withPassword("123456")
        .build()
    ))
    env.execute()
  }

  /**
    * RichSinkFunction:  多了open和close方法  用于打开和关闭连接
    * SinkFunction
    */
  class MysqlSink extends RichSinkFunction[String] {

    var conn: Connection = _

    /**
      * open方法  每一个线程中执行一次  一般用于打开连接
      * 在invoke之前执行
      *
      */
    override def open(parameters: Configuration): Unit = {

      println("创建连接")
      //加载驱动
      Class.forName("com.mysql.jdbc.Driver")
      //创建连接
      conn = DriverManager.getConnection("jdbc:mysql://master:3306/csw?useUnicode=true&characterEncoding=utf-8&useSSL=false", "root", "123456")
    }

    /**
      * invoke:方法，每一条数据都会执行一次
      */
    override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {

      val stat: PreparedStatement = conn.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")

      val split: Array[String] = value.split(",")
      val id: Int = split(0).toInt
      val name: String = split(1)
      val age: Int = split(2).toInt
      val gender: String = split(3)
      val clazz: String = split(4)

      stat.setInt(1, id)
      stat.setString(2, name)
      stat.setInt(3, age)
      stat.setString(4, gender)
      stat.setString(5, clazz)

      stat.executeUpdate()
    }

    /**
      * close方法  每一个线程中执行一次  一般用于关闭连接
      * 在invoke之前执行
      */
    override def close(): Unit = {
      println("关闭连接")
      conn.close()
    }
  }

}
